package org.example.rocketmq.a01_quickstart;

import org.apache.commons.lang3.ThreadUtils;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.example.rocketmq.NamesrvAddr;

import java.time.Duration;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class ProducerAsync {

    public static void main(String[] args) throws Exception {
        asyncProducer();
    }

    /**
     * 不等待消息返回直接进入后续流程， broker 将结果返回后调用 callback 函数，并使用 CountDownLatch 计数。
     */
    public static void asyncProducer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("producer_async_group_01");
        producer.setNamesrvAddr(NamesrvAddr.get());
        producer.setSendMsgTimeout(15000); // 设置超时时间 15秒
        producer.start();

        CountDownLatch countDownLatch = new CountDownLatch(100);
        for (int i = 0; i < 100; i++) {
            final int index = i;
            try {
                String body = ("Hello RocketMQ Async" + index);
                System.out.println(body);
                Message message = new Message("TopicTest", "TagA", body.getBytes(RemotingHelper.DEFAULT_CHARSET));

                producer.send(message, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        countDownLatch.countDown();
                        System.out.printf(index + "消息发送成功 %s%n", sendResult);
                    }

                    @Override
                    public void onException(Throwable throwable) {
                        countDownLatch.countDown();
                        System.out.printf(index + "消息发送是失败 %s%n", throwable.getStackTrace());
                    }
                });

            } catch (Exception e) {
                e.printStackTrace();
                ThreadUtils.sleep(Duration.ofSeconds(1));
            }
        }

        countDownLatch.await(5, TimeUnit.SECONDS);
        producer.shutdown();
    }
}
